﻿using MassTransit;
using MessageBus.Consumer;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Collections.Generic;
using System.Text;
using static MassTransit.HostedServiceConfigurationExtensions;

namespace MessageBus.Extensions
{
    public static class MessageBusExtensions
    {
        /// <summary>
        /// 添加使用ActiveMQ作为消息总线代理
        /// </summary>
        /// <param name="services"></param>
        /// <param name="info">AvtiveMQ连接信息</param>
        /// <param name="receiveKvs">maps<队列,消费者></param>
        /// <returns></returns>
        public static IServiceCollection AddMassTransitByActiveMQ(
            this IServiceCollection services,
            BrokerInfo info,
            IDictionary<string, Action<IReceiveEndpointConfigurator>> receiveKvs = null
            )
        {
            return services.AddMassTransit(x =>
             {
                 x.UsingActiveMq((context, conf) =>
                 {
                     conf.Host(info.IP, info.Port, h =>
                     {
                         h.Username(info.User);
                         h.Password(info.Pwd);
                     });
                     conf.ConfigureEndpoints(context);

                     if (receiveKvs != null)
                         foreach (KeyValuePair<string, Action<IReceiveEndpointConfigurator>> kv in receiveKvs)
                             conf.ReceiveEndpoint(kv.Key, kv.Value);
                 });
                 x.SetKebabCaseEndpointNameFormatter();
             });
        }

        /// <summary>
        /// 添加使用RabbitMQ作为消息总线代理
        /// </summary>
        /// <param name="services"></param>
        /// <param name="info"></param>
        /// <param name="receiveKvs">maps<队列,消费者></param>
        /// <returns></returns>
        public static IServiceCollection AddMassTransitByRabbitMQ(
            this IServiceCollection services,
            BrokerInfo info,
            IDictionary<string, Action<IReceiveEndpointConfigurator>> receiveKvs = null
            )
        {
            return services.AddMassTransit(x =>
            {
                x.UsingRabbitMq((context, conf) =>
                {
                    conf.Host(info.IP, info.Port, info.VirtualHost, c =>
                    {
                        c.Username(info.User);
                        c.Password(info.Pwd);
                    });
                    conf.ConfigureEndpoints(context);

                    if (receiveKvs != null)
                        foreach (KeyValuePair<string, Action<IReceiveEndpointConfigurator>> kv in receiveKvs)
                            conf.ReceiveEndpoint(kv.Key, kv.Value);
                });
                x.SetKebabCaseEndpointNameFormatter();
            });
        }

        public static IServiceCollection AddMassTransitByKafka(
            this IServiceCollection services,
            BrokerInfo info,
            List<MessageConsumer<object>> consumers
            )
        {
            return services.AddMassTransit(x =>
            {
                x.AddRider(rider =>
                {
                    rider.UsingKafka((context, conf) =>
                    {
                        conf.Host($"{info.IP}:{info.Port}");
                        foreach (MessageConsumer<object> consumer in consumers)
                            conf.TopicEndpoint<MessageConsumer<object>>(consumer.Topic, consumer.Group, e =>
                            {
                                e.ConfigureConsumer<MessageConsumer<object>>(context);
                            });
                    });
                });
                x.SetKebabCaseEndpointNameFormatter();
            });
        }

    }
}